package org.zjt.log.exception;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * DESC
 *
 * @author
 * @create 2017-05-23 下午2:30
 *
 * //       hdfs://172.19.110.240:8020/user/catalina.log
 **/
public class LogExceptionJobStart {
    public static void main(String[] args) {
        Configuration config = new Configuration();
        config.set("fs.defaultFS", "hdfs://172.19.110.240:8020");
        config.set("yarn.resourcemanager.hostname", "master");
        System.setProperty("HADOOP_USER_NAME", "root");    //设置用户的角色

//		config.set("mapred.jar", "C:\\Users\\Administrator\\Desktop\\wc.jar");
//		config.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");  设置mapper分词，得到key的策略。
        try {
            FileSystem fs = FileSystem.get(config);

            Job job = Job.getInstance(config);
            job.setJarByClass(LogExceptionJobStart.class);
            job.setJobName("LogExceptionJobStart");

            job.setMapperClass(LogMapper.class);
            job.setReducerClass(LogReducer.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            //partition -> reduce
            //job.setPartitionerClass(WeatherPartition.class);
            /*job.setSortComparatorClass(LogSort.class);*/

            job.setNumReduceTasks(3);

            //job.setInputFormatClass(KeyValueTextInputFormat.class);

            FileInputFormat.addInputPath(job, new Path("/user/catalina.log"));

            Path outpath = new Path("/user/output/log-tmp/");
            if (fs.exists(outpath)) {
                fs.delete(outpath, true);
            }
            FileOutputFormat.setOutputPath(job, outpath);

            boolean f = job.waitForCompletion(true);
            if (f) {
                System.out.println("执行完成");
                Job sortJob = Job.getInstance(config);
                FileInputFormat.addInputPath(sortJob, outpath);

                sortJob.setMapperClass(MyInverseMapper.class);                    //反转 k-V位置
                sortJob.setMapOutputKeyClass(IntWritable.class);
                sortJob.setMapOutputValueClass(Text.class);
                sortJob.setReducerClass(InverseReducer.class);

                sortJob.setOutputKeyClass(Text.class);
                sortJob.setOutputValueClass(IntWritable.class);
                sortJob.setSortComparatorClass(IntWritableComparator.class);
                sortJob.setInputFormatClass(KeyValueTextInputFormat.class);
                sortJob.setNumReduceTasks(1);

                fs.delete(new Path("/user/output/log-result/"),true);
                FileOutputFormat.setOutputPath(sortJob, new Path("/user/output/log-result/"));
                FileInputFormat.addInputPath(sortJob, new Path("/user/output/log-tmp/"));
                sortJob.waitForCompletion(true);
                FileSystem.get(config).delete(new Path("/user/output/log-tmp/"));
                System.exit(0);
            }
            System.out.println("执行完成");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
